/*******************************************************************************
 * Package: com.song.main
 * Type:    TestMap
 * Date:    2024-11-21 19:00
 *
 * Copyright (c) 2024 LTD All Rights Reserved.
 *
 * You may not use this file except in compliance with the License.
 *******************************************************************************/
package com.song.visit;

import com.google.common.collect.Lists;
import org.apache.spark.SparkConf;
import org.apache.spark.api.java.JavaPairRDD;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.api.java.function.PairFunction;
import scala.Tuple2;

import java.util.List;

/**
 *
 * @author Songxianyang
 * @date 2024-11-21 19:00
 */
public class UserVisitMap {

    public static void main(String[] args) {
        // 1.创建配置对象
        SparkConf conf = new SparkConf().setMaster("local").setAppName("sparkCore");

        // 2. 创建sparkContext
        JavaSparkContext sc = new JavaSparkContext(conf);

        List<Integer> data = Lists.newArrayList(1, 2, 3, 4);

        JavaRDD<String> lines = sc.textFile("yqmm-spark/file/user_visit_action.txt");
        // 行 过滤掉无用数据  （搜索）
        JavaRDD<String> filterRdd = lines.filter(row -> {

                    String[] s = row.split("_");
                    return "null".equals(s[5]);
                }
        );
        // 点击 找到
        // click 找到
        JavaRDD<String> click = filterRdd.filter(row -> {

                    String[] s = row.split("_");
                    return !"-1".equals(s[6]);
                }
        );
        // 聚合
        JavaPairRDD<String, Integer> resp = click.mapToPair(new PairFunction<String, String, Integer>() {
            private static final long serialVersionUID = -2300816024782909714L;

            @Override
            public Tuple2<String, Integer> call(String row) throws Exception {
                String[] s = row.split("_");
                return new Tuple2<>(s[6], 1);
            }
        }).reduceByKey(Integer::sum);

        System.out.println(resp.take(5));

        sc.close();
    }
}
